﻿using WinDocker.Common.Global;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
using MQTTnet.Client.Receiving;
using MQTTnet.Client;
using MQTTnet.Protocol;
using MQTTnet;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace WinDocker.Common.Helper.Mqtt
{
    /// <summary>
    ///  MQTT处理对像
    /// </summary>
    public class MQTTHelper
    {
        /// <summary>
        /// 单例模式对像获取
        /// </summary>
        public static readonly MQTTHelper Instance = new MQTTHelper();
        /// <summary>
        /// MQTT数据接收事件
        /// </summary>
        public event EventHandler MQTTReceivedDataEvent;
        /// <summary>
        /// MQTT客户端连接服务端成功事件
        /// </summary>
        public event EventHandler MQTTConnectedEvent;
        /// <summary>
        /// MQTT客户端与服务端连接断开事件
        /// </summary>
        public event EventHandler MQTTDisconnectedEvent;

        /// <summary>
        /// MQTT客户端重新连接服务端失败事件
        /// </summary>
        public event EventHandler MQTTReConnectedFailEvent;

        /// <summary>
        /// 是否启用顺序发送接收处理
        /// </summary>
        public bool IsOrder { get; set; } = true;

        private Dictionary<string, MQTTReceivedDataTopicOrderModel> _DicSendReceiveTopicOrder = new Dictionary<string, MQTTReceivedDataTopicOrderModel>();
        /// <summary>
        /// 顺序发送接收主题(Topic)哈希列表
        ///   为空表示无需处理
        ///   《发送的主题(Topic),MQTT数据顺序接收处理配置实体类对像(内含接收主题[Topic]信息)》
        /// </summary>
        public Dictionary<string, MQTTReceivedDataTopicOrderModel> DicSendReceiveTopicOrder
        {
            get
            {
                return _DicSendReceiveTopicOrder;
            }
        }
        /// <summary>
        /// 顺序发送超时秒数
        /// </summary>
        public int OrderSendTimeOutSeconds
        {
            get; set;
        } = 5;
        /// <summary>
        /// MQTT发布数据实体类对像的哈希表
        /// 《发送主题(Topic), MQTT发布数据实体类对像的队列》
        /// </summary>
        private Dictionary<string, Queue<MQTTPublishDataModel>> DicMQTTPublishDataModel = new Dictionary<string, Queue<MQTTPublishDataModel>>();
        /// <summary>
        /// 数据顺序发送接收时,是否启用缓存 即在发送发布无订阅接收应答反馈时，对发送发布的数据进行缓存,以备通讯正常后恢复发送.
        /// </summary>
        public bool IsOrderCache { get; set; } = false;

        private MQTTHelper()
        {
            try
            {
                GlobalParameters.Init();

                client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e =>
                {
                    //ShowMsg("### RECEIVED APPLICATION MESSAGE ###");
                    //ShowMsg($"+ Topic = {e.ApplicationMessage.Topic}");
                    //ShowMsg($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
                    //ShowMsg($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}");
                    //ShowMsg($"+ Retain = {e.ApplicationMessage.Retain}");                     
                    try
                    {
                        MQTTReceivedDataModel mQTTReceivedDataModel = new MQTTReceivedDataModel()
                        {
                            Topic = e.ApplicationMessage.Topic,
                            Payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload),
                            QoS = e.ApplicationMessage.QualityOfServiceLevel,
                            Retain = e.ApplicationMessage.Retain,
                            ClientId = e.ClientId
                        };
                        this.DicSendReceiveTopicOrderSet(mQTTReceivedDataModel);
                        this.MQTTReceivedDataEvent?.Invoke(mQTTReceivedDataModel, null);
                    }
                    catch (Exception ex)
                    {
                        LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper 1Err:" + ex.Message);
                    }
                });

                //((MqttClient)client).ApplicationMessageReceived += ((s, e) => { });

                client.ConnectedHandler = new MqttClientConnectedHandlerDelegate(async e =>
                {
                    //ShowMsg("### CONNECTED WITH SERVER ###");

                    ////await client.SubscribeAsync(new TopicFilterBuilder().WithTopic("#").Build());

                    //ShowMsg("### SUBSCRIBED ###");
                    try
                    {
                        this.MQTTConnectedEvent?.Invoke("连接MQTT服务端成功!", null);
                    }
                    catch (Exception ex)
                    {
                        LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper 2Err:" + ex.Message);
                    }
                });

                client.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(async e =>
                {
                    //ShowMsg("### DISCONNECTED FROM SERVER ###");                   
                    try
                    {
                        this.MQTTDisconnectedEvent?.Invoke("MQTT客户端与服务端的连接断开了!!", null);
                        await Task.Delay(TimeSpan.FromSeconds(5));
                        try
                        {
                            await client.ConnectAsync(clientOptions);
                            LogWrite.TxtExport_DateNow("LogTrace.txt", "\r\n【对Mqtt Broker进行重连操作】\r\n");
                        }
                        catch (Exception ex)
                        {
                            //ShowMsg("### RECONNECTING FAILED ###");
                            this.MQTTReConnectedFailEvent?.Invoke("MQTT客户端重新连接服务端失败了!!" + ex.Message, null);
                        }
                    }
                    catch (Exception ex)
                    {
                        LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper 3Err:" + ex.Message);
                    }
                });
            }
            catch (Exception ex)
            {
                LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper Err:" + ex.Message);
            }
        }

        private void DicSendReceiveTopicOrderSet(MQTTReceivedDataModel mQTTReceivedDataModel)
        {
            try
            {
                if (!this.IsOrder)
                {
                    return;
                }
                foreach (var v in this.DicSendReceiveTopicOrder)
                {
                    var v2 = v.Value;
                    if (v2.Topic == mQTTReceivedDataModel.Topic)
                    {
                        v2.IsSend = true;
                        this.CacheListPublish(v.Key);
                    }
                }
            }
            catch (Exception ex)
            {
                LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper DicSendReceiveTopicOrderSet Err:" + ex.Message);
            }
        }
        /// <summary>
        /// 缓存列表发布处理
        /// </summary>
        /// <param name="publishTopic"></param>
        private void CacheListPublish(string publishTopic)
        {
            try
            {
                if (!this.IsOrder)
                {
                    return;
                }
                if (!this.DicMQTTPublishDataModel.ContainsKey(publishTopic))
                {
                    return;
                }
                var vQueue = this.DicMQTTPublishDataModel[publishTopic];
                while (true)
                {
                    if (vQueue.Count <= 0)
                    {
                        break;
                    }
                    var v = vQueue.Dequeue();
                    this.Publish(v.topic, v.msg, v.iPublishQos);
                }
            }
            catch (Exception ex)
            {
                LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper CacheListPublish Err:" + ex.Message);
            }
        }

        IMqttClient client = new MqttFactory().CreateMqttClient();
        IMqttClientOptions clientOptions = new MqttClientOptions();
        /// <summary>
        /// 连接0
        /// </summary>
        /// <param name="serverAddr">MqttBroker服务器地址(如:47.110.46.216)</param>
        /// <param name="iPort">MqttBroker服务器端口(如:1883)</param>
        public void Connect(string serverAddr, int iPort, string user = "", string pwd = "", string clientId = "")
        {
            try
            {
                user = (user ?? "").Trim();
                //LogWrite.TxtExport_DateNow("LogTest.txt", $"Dx:{serverAddr},{iPort}");
                this.Connect(new MqttClientTcpOptions() { Server = serverAddr, Port = iPort }, (user == "" ? null : new MqttClientCredentials() { Username = user, Password = Encoding.Default.GetBytes(pwd) }), clientId);
            }
            catch (Exception ex)
            {
                LogWrite.TxtExport_DateNow("LogErr.txt", $"MQTTHelper Connect Err:{ex.Message}");
            }
        }

        /// <summary>
        /// 连接
        /// </summary>
        /// <returns></returns>
        public void Connect(MqttClientTcpOptions mqttClientTcpOptions = null, MqttClientCredentials mqttClientCredentials = null, string clientId = "")
        {
            try
            {
                clientId = (clientId ?? "").Trim();
                //var will = new MqttApplicationMessage() { Topic = "lastWill", Payload = Encoding.UTF8.GetBytes("掉线了!") };//定义遗嘱消息
                clientOptions = new MqttClientOptions
                {
                    ChannelOptions = ((mqttClientTcpOptions == null) ? (new MqttClientTcpOptions
                    {
                        Server = GlobalParameters.MQTTServer,//txtIP.Text
                        Port = GlobalParameters.MQTTPort
                    }) : mqttClientTcpOptions),
                    Credentials = ((mqttClientCredentials == null) ? (new MqttClientCredentials
                    {
                        Username = GlobalParameters.MQTTUserName,//"admin",
                        Password = Encoding.Default.GetBytes(/*"123456"*/GlobalParameters.MQTTPassWord)
                    }) : mqttClientCredentials),
                    ClientId =
                    (
                    GlobalParameters.IsBinJiang ?
                    GlobalParameters.MQTTClientId :
                    (clientId == "" ? clientOptions.ClientId : clientId)
                    ),
                    WillMessage = new MqttApplicationMessage() { Topic = "lastwill", Payload = Encoding.UTF8.GetBytes("掉线了!") }
                };

                /*var  v =*/
                client.ConnectAsync(clientOptions);
                //MqttClientAuthenticateResult vv = v.Result;                
            }
            catch (Exception ex)
            {
                LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper Conn Err:" + ex.Message);
            }
            //return false;
        }

        public void DisConnect()
        {
            try
            {
                if (client != null)
                {
                    client.DisconnectAsync();
                }
            }
            catch (Exception ex)
            {
                LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper DisConn Err:" + ex.Message);
            }
        }

        /// <summary>
        /// 订阅
        /// </summary>
        /// <param name="topic">订阅主题</param>
        /// <param name="iSubscribeQos"></param>
        public void Subscribe(string topic, int iSubscribeQos = 0)
        {
            try
            {
                MqttQualityOfServiceLevel mqttQualityOfServiceLevel = new MqttQualityOfServiceLevel();
                switch (/*cboSubscribeQos.SelectedIndex*/iSubscribeQos)
                {
                    case 0:
                        mqttQualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
                        break;
                    case 1:
                        mqttQualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce;
                        break;
                    case 2:
                        mqttQualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce;
                        break;
                    default:
                        mqttQualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
                        break;
                }
                client.SubscribeAsync(new TopicFilter { Topic = topic/*txtSubscribe.Text*/, QualityOfServiceLevel = mqttQualityOfServiceLevel });
                //try
                //{
                //    var will = new MqttApplicationMessage() { Topic = topic, Payload = Encoding.UTF8.GetBytes(topic+ " 掉线了!") };//定义遗嘱消息

                //}
                //catch (Exception ex)
                //{
                //    LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper Subscribe 0Err:" + ex.Message);
                //}
            }
            catch (Exception ex)
            {
                LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper Subscribe Err:" + ex.Message);
            }
        }

        /// <summary>
        /// 发布
        /// </summary>
        /// <param name="topic">发布主题</param>
        /// <param name="msg">发布内容</param>
        /// <param name="iPublishQos"></param>
        public void Publish(string topic, string msg, int iPublishQos = 0)
        {
            try
            {
                MqttQualityOfServiceLevel mqttQualityOfServiceLevel = new MqttQualityOfServiceLevel();
                switch (/*cboPublishQos.SelectedIndex*/iPublishQos)
                {
                    case 0:
                        mqttQualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
                        break;
                    case 1:
                        mqttQualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce;
                        break;
                    case 2:
                        mqttQualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce;
                        break;
                    default:
                        mqttQualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
                        break;
                }

                var applicationMessage = new MqttApplicationMessageBuilder()
                            .WithTopic(/*txtTopic.Text*/topic)
                            .WithPayload(msg/*"Hello World" + DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss fff") + " -- " + txtPublish.Text*/)
                            //.WithAtLeastOnceQoS()
                            //.WithAtLeastOnceQoS()
                            .Build();
                applicationMessage.QualityOfServiceLevel = mqttQualityOfServiceLevel;
                if (this.GetPublishJudge(topic))
                {
                    client.PublishAsync(applicationMessage);
                    this.DicSendReceiveTopicOrderSendSet(topic);
                }
                else
                {
                    if (!this.IsOrderCache)
                    {
                        return;
                    }
                    var mQTTPublishDataModel = new MQTTPublishDataModel() { topic = topic, msg = msg, iPublishQos = iPublishQos };
                    if (this.DicMQTTPublishDataModel.ContainsKey(topic))
                    {
                        this.DicMQTTPublishDataModel[topic].Enqueue(mQTTPublishDataModel);
                    }
                    else
                    {
                        var vQueue = new Queue<MQTTPublishDataModel>();
                        vQueue.Enqueue(mQTTPublishDataModel);
                        this.DicMQTTPublishDataModel.Add(topic, vQueue);
                    }
                }
            }
            catch (Exception ex)
            {
                LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper Publish Err:" + ex.Message);
            }
        }
        /// <summary>
        /// 获取是否允许进行发布操作
        /// </summary>
        /// <param name="topic"></param>
        /// <returns></returns>
        private bool GetPublishJudge(string topic)
        {
            try
            {
                if (!this.IsOrder)
                {
                    return true;
                }
                foreach (var v in this.DicSendReceiveTopicOrder)
                {
                    if (v.Key == topic)
                    {
                        return v.Value.IsSend;
                    }
                }
            }
            catch (Exception ex)
            {
                LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper GetPublishJudge Err:" + ex.Message);
            }
            return true;
        }

        /// <summary>
        /// 发布同步设置
        /// </summary>
        /// <param name="topic"></param>
        private void DicSendReceiveTopicOrderSendSet(string topic)
        {
            try
            {
                if (!this.IsOrder)
                {
                    return;
                }
                foreach (var v in this.DicSendReceiveTopicOrder)
                {
                    if (v.Key == topic)
                    {
                        var v2 = v.Value;
                        v2.IsSend = false;
                    }
                }
                Thread th = new Thread(this.DicSendReceiveTopicOrderSendSetTimeOut);
                th.Start(topic);
            }
            catch (Exception ex)
            {
                LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper DicSendReceiveTopicOrderSendSet Err:" + ex.Message);
            }
        }

        private readonly object DicSendReceiveTopicOrderSendSetTimeOutLockObj = new object();
        /// <summary>
        /// 发布同步超时处理
        /// </summary>
        /// <param name="obj"></param>
        private void DicSendReceiveTopicOrderSendSetTimeOut(object obj)
        {
            string topic = "";
            try
            {
                if (!this.IsOrder)
                {
                    return;
                }
                topic = (string)obj;
            }
            catch { }
            object objLock = DicSendReceiveTopicOrderSendSetTimeOutLockObj;
            try
            {
                foreach (var v in this.DicSendReceiveTopicOrder)
                {
                    if (v.Key == topic)
                    {
                        objLock = v.Value.LockObj;
                        break;
                    }
                }
            }
            catch (Exception ex)
            {
                LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper DicSendReceiveTopicOrderSendSetTimeOut(获取锁) Err:" + ex.Message);
            }
            lock (objLock)
            {
                try
                {
                    for (int i = 0; i < this.OrderSendTimeOutSeconds; i++)
                    {
                        Thread.Sleep(1000);
                    }
                    foreach (var v in this.DicSendReceiveTopicOrder)
                    {
                        if (v.Key == topic)
                        {
                            var v2 = v.Value;
                            //v2.IsSend = false;
                            if (!v2.IsSend)
                            {
                                v2.IsSend = true;
                            }
                            break;
                        }
                    }
                }
                catch (Exception ex)
                {
                    LogWrite.TxtExport_DateNow("LogErr.txt", "MQTTHelper DicSendReceiveTopicOrderSendSetTimeOut Err:" + ex.Message);
                }
            }
        }

    }

    /// <summary>
    /// MQTT数据接收实体类
    /// </summary>
    public class MQTTReceivedDataModel
    {
        /// <summary>
        /// 主题
        /// </summary>
        public string Topic { get; set; }
        /// <summary>
        /// (有效载荷)接收到的数据内容
        /// </summary>
        public string Payload { get; set; }

        public MqttQualityOfServiceLevel QoS { get; set; }

        public bool Retain { get; set; }

        public string ClientId { get; set; }

        public DateTime CreateDt { get; set; } = DateTime.Now;
    }

    /// <summary>
    /// MQTT数据顺序接收处理配置实体类
    /// </summary>
    public class MQTTReceivedDataTopicOrderModel
    {
        /// <summary>
        /// 接收主题名
        /// </summary>
        public string Topic { get; set; }
        /// <summary>
        /// 能否进行下一条数据的发送
        /// </summary>
        public bool IsSend { get; set; } = true;

        private readonly object _LockObj = new object();
        /// <summary>
        /// 锁对像
        /// </summary>
        public object LockObj { get { return _LockObj; } }
    }

    /// <summary>
    /// MQTT发布数据实体类
    /// </summary>
    class MQTTPublishDataModel
    {
        public string topic { get; set; }

        public string msg { get; set; }

        public int iPublishQos { get; set; }
    }
}
